<?php

declare(strict_types=1);

namespace App\CustHub\Process;

use App\CustHub\Service\EasyWorkNotifyService;
use Carbon\Carbon;
use Hyperf\Contract\StdoutLoggerInterface;
use Hyperf\Coroutine\Coroutine;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Nashgao\MQTT\Utils\Qos;
use Simps\MQTT\Client;
use Simps\MQTT\Config\ClientConfig;
use Simps\MQTT\Exception\ConnectException;
use Simps\MQTT\Protocol\ProtocolInterface;
use Simps\MQTT\Protocol\Types;
use Throwable;
use function Hyperf\Coroutine\co;

#[Process(name: 'MqttSubscriber', enableCoroutine: true)]
class MqttSubscriberProcess extends AbstractProcess
{
    // 使用连接池，避免重复创建连接
    private ?Client $client = null;

    public function handle(): void
    {
        stdout_log()->info('MQTT订阅开始...');
        $logger = $this->container->get(StdoutLoggerInterface::class);

        while (true) {
            try {
                $this->connectAndSubscribe($logger);
            } catch (Throwable $e) {
                $logger->error("MQTT订阅异常: " . $e->getMessage());
                $this->reconnect($logger);
            }
        }
    }

    private function connectAndSubscribe(StdoutLoggerInterface $logger): void
    {
        // 从配置文件获取MQTT参数
        $config = config('mqtt.default', []);
        $host = $config['host'] ?? 'localhost';
        $port = $config['port'] ?? 1883;
        $username = $config['username'] ?? '';
        $password = $config['password'] ?? '';

        // 为共享订阅生成统一的客户端ID前缀，确保同一组服务器使用相同的组名
        $groupId = $config['share_group'] ?? 'server-group'; // 共享组名，多台服务器保持一致
        $clientId = $config['client_id'] ?? 'hyperf_subscriber_' . $groupId . uniqid();

        $swooleConfig = $config['swoole_config'];

        $config = new ClientConfig();
        $config->setClientId($clientId);
        $config->setUsername($username);
        $config->setPassword($password);
        $config->setKeepAlive(60);
        $config->setDelay(3000);
        $config->setProtocolLevel(ProtocolInterface::MQTT_PROTOCOL_LEVEL_5_0);
        $config->setSwooleConfig($swooleConfig);

        $this->client = make(Client::class, [$host, (int)$port, $config]);

        while (!$this->client->connect(false)) {
            \Swoole\Coroutine::sleep(3);
            stdout_log()->info('mqtt3s后重连...');
            $this->client->connect(false);
        }

        //MQTT共享订阅 消息会在服务器间进行负载均衡，每条消息只会被一台服务器处理。

        $topicsKey = env('APP_ENV') != 'dev' ? '$share/' . $groupId . '/wework/msg/receive' : 'wework/msg/receive';

        $topics = [
            $topicsKey => [
                'qos' => 1,
                'no_local' => true,
                'retain_as_published' => true,
                'retain_handling' => 2,
            ],
        ];

        $this->client->subscribe($topics);
        $timeSincePing = time();
        $lastMessageTime = time();

        while (true) {
            $buffer = $this->client->recv();
            if ($buffer && $buffer !== true) {
                $lastMessageTime = time();
                if ($buffer['type'] === Types::PUBLISH && $buffer['qos'] === Qos::QOS_AT_LEAST_ONCE) {
                    try {
                        $this->client->send([
                            'type' => Types::PUBACK,
                            'message_id' => $buffer['message_id']
                        ], false);
                    } catch (Throwable $e) {
                        $logger->error('PUBACK发送失败: ' . $e->getMessage());
                    }
                }

                if ($buffer['type'] === Types::PUBLISH) {
                    $this->handleMessage($buffer, $logger);
                }

            }

            $keepAlive = $config->getKeepAlive();
            $timeSinceLastPing = time() - $timeSincePing;
            // 如果距离上次心跳已经超过了保活时间，需要发送心跳
            if ($timeSinceLastPing >= $keepAlive) {
                if ($this->client->ping()) {
                    $timeSincePing = time();
                    stdout_log()->info('发送心跳包成功'.Carbon::now()->toDateTimeString());
                } else {
                    $this->client->close();
                    throw new ConnectException('MQTT服务器心跳发送失败，连接断开');
                }
            }

            // 检查长时间无消息
            if (time() - $lastMessageTime > 300) {
                $logger->warning('超过5分钟无消息，重连...');
                $this->client->close();
                throw new ConnectException('MQTT连接长时间无消息');
            }
        }
    }

    private function handleMessage($message, StdoutLoggerInterface $logger): void
    {
        co(function () use ($message, $logger) {
            // 根据你的业务逻辑处理消息
            $topic = $message['topic'] ?? 'unknown';
            $payload = $message['message'] ?? '';

            stdout_log()->info("收到MQTT消息" . json_encode([
                    'topic' => $topic,
                    'payload' => $payload,
                    'time' => date('Y-m-d H:i:s')
                ]));

            try {
                // 示例业务逻辑 - 根据主题分发处理
                switch (true) {
                    case str_contains($topic, 'wework/msg/receive'):
                        $this->processSensorData($payload);
                        break;
                    default:
                        $logger->warning("未处理的主题: " . $topic);
                }
            } catch (Throwable $e) {
                $logger->error("业务处理失败: " . $e->getMessage());
                // 根据业务决定是否重试或记录错误
            }
        });
    }

    private function processSensorData(string $payload): void
    {
        // 处理传感器数据
        $params = json_decode($payload, true);
        if (isset($params['code']) && $params['code'] == 0) {
            $data = $params['data'];
            $msgTypes = array_column($data, 'msgType');
            if (count(array_unique($msgTypes)) > 1) {
                $data = array_filter($data, function ($item) {
                    return $item['msgType'] != 2118;
                });
                // 重建索引
                $data = array_values($data);
            }

            foreach ($data as $item) {
                co(function () use ($item) {
                    $item = camelToUnder($item);
                    $this->container->get(EasyWorkNotifyService::class)->notify($item);
                });
            }
        }
    }


    /**
     * 从共享订阅主题中提取原始主题名
     */
    private function extractOriginalTopic(string $sharedTopic): string
    {
        // 共享订阅主题格式: $share/group/original/topic
        $parts = explode('/', $sharedTopic);

        // 移除前两部分 ($share 和 group名)
        if (count($parts) >= 3 && $parts[0] === '$share') {
            array_shift($parts); // 移除 $share
            array_shift($parts); // 移除 group名
            return implode('/', $parts);
        }

        return $sharedTopic;
    }

    private function reconnect(StdoutLoggerInterface $logger): void
    {
        if ($this->client) {
            try {
                $this->client->close();
            } catch (Throwable $e) {
                $logger->error("关闭MQTT连接异常: " . $e->getMessage());
            }
            $this->client = null;
        }

        $logger->info("等待5秒后重连...");
        Coroutine::sleep(5);
    }

    public function isEnable($server): bool
    {
        return \Hyperf\Support\env('MQTT_ENABLE', true);
    }
}